package cn.tedu.charging.order.emqx;

import cn.tedu.charging.common.constant.MqttDestinationConst;
import cn.tedu.charging.common.pojo.entity.ProgressEntity;
import cn.tedu.charging.common.pojo.message.ChargingCheckResultMessage;
import cn.tedu.charging.common.pojo.message.ChargingProgressMsg;
import cn.tedu.charging.common.pojo.message.ProgressData;
import cn.tedu.charging.common.pojo.param.ProgressCostParam;
import cn.tedu.charging.common.pojo.po.ChargingBillFailPO;
import cn.tedu.charging.common.pojo.po.ChargingBillSuccessPO;
import cn.tedu.charging.common.pojo.vo.ProgressCostVO;
import cn.tedu.charging.common.protocol.WebSocketMessage;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.springframework.beans.BeanUtils;
import org.springframework.data.redis.core.ValueOperations;

import java.nio.charset.StandardCharsets;
import java.util.Date;
import java.util.concurrent.TimeUnit;

@Slf4j
public class MqttCallback implements MqttCallbackExtended {

    private final MqttClient mqttClient;
    private final MqttContext mqttContext;

    public MqttCallback(MqttClient mqttClient, MqttContext mqttContext){
        this.mqttClient = mqttClient;
        this.mqttContext = mqttContext;
    }

    public void connectComplete(boolean reconnect, String serverURI) {
        if(reconnect){
            log.info("自动重连成功,borker服务端地址:{}",serverURI);
        }else{
            log.info("首次连接成功,borker服务端地址:{}",serverURI);
        }
        try{
            // 共享订阅 设备反馈自检结果
            mqttClient.subscribe("$share/order/" + MqttDestinationConst.GUN_CHECK_RESULT_TOPIC,1);
            // 共享订阅 订阅设备充电进度
            mqttClient.subscribe("$share/order/" + MqttDestinationConst.CHARGING_PROGRESS_TOPIC,1);
        } catch (Exception e){
            log.error("订阅主题失败",e);
        }
    }

    public void connectionLost(Throwable throwable) {
        log.info("连接断开,原因:{}",throwable.getMessage());
    }

    public void messageArrived(String topic, MqttMessage message) {
        log.info("收到消息,来源topic:{},消息内容:{}",topic, new String(message.getPayload(), StandardCharsets.UTF_8));
        // 判断解析消息 设备反馈消息
        if(MqttDestinationConst.GUN_CHECK_RESULT_TOPIC.equals(topic)){
            // 对消息做反序列化
            MySerializer<ChargingCheckResultMessage> serializer = mqttContext.getMySerializer();
            ChargingCheckResultMessage checkMessage = serializer.deserialize(message.getPayload(),
                    ChargingCheckResultMessage.class);
            // 添加分布式锁
            String lockKey = "charging:bill:lock:" + checkMessage.getOrderNo();
            String value = "emqx of key";
            ValueOperations<String, String> ops = mqttContext.getRedisTemplate().opsForValue();
            boolean lock; int count = 0;
            try {
                do {
                    count++;
                    if (count > 1) {
                        Thread.sleep(1000);
                    }
                    lock = Boolean.TRUE.equals(ops.setIfAbsent(lockKey, value, 5, TimeUnit.MINUTES));
                } while(!lock);
            } catch (InterruptedException e) {
                log.error("抢分布式锁失败",e);
            }
            // 调用业务方法处理
            handleCheckResult(checkMessage);
            // 释放锁
            mqttContext.getRedisTemplate().delete(lockKey);
        } else if(MqttDestinationConst.CHARGING_PROGRESS_TOPIC.equals(topic)){
            MySerializer<ChargingProgressMsg> serializer = mqttContext.getMySerializer();
            ChargingProgressMsg cpm = serializer.deserialize(message.getPayload(),
                    ChargingProgressMsg.class);
            handleProgressing(cpm);
        }
    }

    private void handleCheckResult(ChargingCheckResultMessage checkMessage) {
        // 1.利用订单编号orderNo  billId 查询存在订单 success|fail
        String billID = checkMessage.getOrderNo();
        QueryWrapper queryWrapper = new QueryWrapper();
        queryWrapper.eq("bill_id",billID);
        Long success = mqttContext.getBillSuccessMapper().selectCount(queryWrapper);
        Long fail = mqttContext.getBillFailMapper().selectCount(queryWrapper);
        // 2.判断是否存在 存在则结束 不存在下一步 只要上述两个count有一个>0 则结束
        if(success > 0 || fail > 0){
            log.warn("订单:{}重复消费",billID);
            return;
        }
        boolean result = checkMessage.getResult();
        if (result) {
            // 3.1 true 组织successPO写入success表格 发送命令开始充电(undo)
            log.debug("订单:{},设备自检结果成功", billID);
            ChargingBillSuccessPO po = new ChargingBillSuccessPO();
            // 订单编号 user用户身份 创建时间 修改时间 订单刚创建初始状态 1=订单正在充电中
            po.setBillStatus(1);
            po.setBillId(billID);
            po.setUserId(checkMessage.getUserId());
            po.setCreateTime(new Date());
            // 缺少的字段 考虑重新定义接口 让设备传递的消息 包含更多属性
            mqttContext.getBillSuccessMapper().insert(po);
            // 设备自检成功 订单创建成功后 给设备发送第二个命令 开始给车充电
            mqttContext.getDeviceClient().changeGunStatus(checkMessage.getGunId(),2);
        } else {
            log.debug("订单:{},设备自检结果失败", billID);
            // 3.2 false 组织failPO写入fail表格,调用设备接口修改枪状态(undo) 实时消息推送(TODO)
            ChargingBillFailPO po = new ChargingBillFailPO();
            // 订单编号 用户id 时间
            po.setBillId(billID);
            po.setUserId(checkMessage.getUserId());
            po.setCreateTime(new Date());
            po.setFailDesc("设备自检失败");
            mqttContext.getBillFailMapper().insert(po);
            // 设计一个修改枪状态的device服务接口
            mqttContext.getDeviceClient().changeGunStatus(checkMessage.getGunId(),3);
            // 实时消息推送 告知设备故障 或者其他原因 订单无法继续充电 请换枪 赠送优惠券
            // 订单失败 组织WebSocketMessage 给用户推送消息
            WebSocketMessage<String> wsm = new WebSocketMessage<>();
            // state=1表示订单有问题 2用户余额不足 3正常同步充电进度
            wsm.setState(1);
            wsm.setMessage("订单失败 请换枪");
            wsm.setData("B扫码下单失败，请换枪，非常抱歉，赠送您一张优惠券:http：//AN0df32fj3");
            mqttContext.getWebSocketServer().sendMessage(checkMessage.getUserId(),wsm);

        }
    }

    private void handleProgressing(ChargingProgressMsg chargingProgressMsg) {
        // 1.远程调用cost服务,获取本次充电进度计价结果 得到总价和总时间
        ProgressCostParam costParam = new ProgressCostParam();
        BeanUtils.copyProperties(chargingProgressMsg,costParam);
        ProgressCostVO vo = mqttContext.getCostClient().calculateCost(costParam).getData();
        // 2.将充电进度存储到es
        ProgressEntity pe = new ProgressEntity();
        BeanUtils.copyProperties(chargingProgressMsg,pe);
        pe.setTotalCost(vo.getTotalCost());
        // id补充 雪花算法 保证订单充电进度 时序正常的
        pe.setId(String.valueOf(mqttContext.getSnowflakeIdGenerator().nextId()));
        mqttContext.getEsRepository().save(pe);
        // 3.根据充电进度 充电是否满了,判断给用户推送什么消息
        if (chargingProgressMsg.getIsFull()){
            //按照弹窗处理
            WebSocketMessage<String> message=new WebSocketMessage<>();
            message.setState(1);//失败异常
            message.setMessage("充满");
            message.setData("A已充满,请结束订单,支付");
            mqttContext.getWebSocketServer().sendMessage(chargingProgressMsg.getUserId(),message);
        } else {
            // 假设 当前充电度数 16度 计算总充电量100 16%
            // 当前业务中 有专门提示用户充电进度的实时数据封装
            WebSocketMessage<ProgressData> message = new WebSocketMessage<>();
            message.setState(3);// state=3表示充电进度展示
            message.setMessage("充电中");
            ProgressData data = new ProgressData();
            data.setTotalCapacity(100.00); // 总电量
            data.setTotalCost(vo.getTotalCost()); // 已产生的总费用
            data.setOneElectricityCost(vo.getPowerFee()); // 单价

            Long totalSeconds = vo.getTotalTime(); // 总秒数
            int hours = Math.toIntExact(totalSeconds / 3600); // 小时
            int minutes = Math.toIntExact((totalSeconds % 3600) / 60); // 分钟
            int seconds = Math.toIntExact(totalSeconds % 60); // 秒

            data.setHours(hours);
            data.setMinutes(minutes);
            data.setSeconds(seconds);

            Double chargingCapacity = chargingProgressMsg.getChargingCapacity();
            String totalCapacityKey = "charging:total:capacity:" + chargingProgressMsg.getOrderNo();
            Double totalCap = mqttContext.getRedisTemplate().opsForValue().increment(totalCapacityKey,chargingCapacity);
            data.setChargingCapacity(totalCap);

            message.setData(data);
            mqttContext.getWebSocketServer().sendMessage(chargingProgressMsg.getUserId(),message);
        }
    }

    public void deliveryComplete(IMqttDeliveryToken token) {
        // 可以从token获得发送的原始消息
        String body = null;
        try{
            MqttMessage message = token.getMessage();
            body = new String(message.getPayload(),StandardCharsets.UTF_8);
        }catch (Exception e){
            log.error("获取原始消息失败",e);
        }
        log.info("消息投递完成,消息ID:{},消息原始内容:{}",token.getMessageId(),body);
    }
}
